package com.test.usercenter;

import com.jcraft.jsch.IO;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

public class Relations {
    private Configuration conf = null;
    private Connection conn = null;

    @Before
    public void init() {
        try {
            //获取一个配置文件对象
            conf = HBaseConfiguration.create();

            conf.set("hbase.zookeeper.quorum", "tctj1,tctj2");
            conf.set("hbase.zookeeper.property.clientPort", "2181");
            //通过conf获取到hbase集群的连接
            conn = ConnectionFactory.createConnection(conf);
            // create table
            createTable();
            truncateTable();
            // 添加好友
            addFriend("id1","id2");
            addFriend("id1","id5");
            addFriend("id1","id7");
            addFriend("id2","id1");
            addFriend("id2","id3");
            addFriend("id2","id6");
            addFriend("id2","id10");
            addFriend("id3","id2");
            addFriend("id5","id1");
            addFriend("id6","id2");
            addFriend("id7","id1");
            addFriend("id10","id2");
            // 获取数据
            getData();
        } catch (IOException ex) {
            System.out.println(ex);
        }
    }

    /**
     * 创建好友表
     */
    public void createTable() {
        try {
            //获取HbaseAdmin对象用来创建表
            HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
            if (admin.tableExists(Constants.TABLE_RELATION)) {
                return;
            }
            //创建Htabledesc描述器，表描述器
            final HTableDescriptor table = new HTableDescriptor(TableName.valueOf(Constants.TABLE_RELATION));
            //指定列族
            table.addFamily(new HColumnDescriptor(Constants.COL_FRIEND));
            admin.createTable(table);
            System.out.println(Constants.TABLE_RELATION + " - 表创建成功！！");
        } catch (IOException ex) {
            System.out.println(ex);
        }
    }

    public void truncateTable() {
        try {
            HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
            System.out.println("开始清空数据");
            //取得目标数据表的表明对象
            TableName tableName = TableName.valueOf(Constants.TABLE_RELATION);
            //设置表状态为无效
            admin.disableTable(tableName);
            //清空指定表的数据
            admin.truncateTable(tableName, true);
            System.out.println("结束");
        } catch (IOException e) {
            e.printStackTrace();
        }


    }
    /**
     * 添加朋友
     */
    public void addFriend(String fid, String mid) {
        try {
            //需要获取一个table对象
            final Table table = conn.getTable(TableName.valueOf(Constants.TABLE_RELATION));
            //准备put对象
            final Put put = new Put(Bytes.toBytes(fid));    //指定rowkey
            put.addColumn(Bytes.toBytes(Constants.COL_FRIEND), Bytes.toBytes(mid), Bytes.toBytes(mid));
            table.put(put);
            table.close();

        } catch (IOException ex) {
            System.out.println(ex);
        }
    }

    public void getData() {
        try  {
            final Table table = conn.getTable(TableName.valueOf(Constants.TABLE_RELATION));
            //准备scan对象
            final Scan scan = new Scan();
            //执行扫描
            final ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                //获取到result中所有cell对象
                final Cell[] cells = result.rawCells();
                //遍历打印
                for (Cell cell : cells) {
                    final String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    final String f = Bytes.toString(CellUtil.cloneFamily(cell));
                    final String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                    final String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("rowkey-->" + rowkey + "--;cf-->" + f + ";column--->" + column + "--;value-->" + value);
                }
            }

            table.close();
        } catch (IOException ex) {
            System.out.println(ex);
        }
    }


    //释放连接
    @After
    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
